package com.atguigu.spark.util.req

import com.atguigu.spark.util.util.{DruidUtil, PropUtil}
import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Duration, StreamingContext}

import java.sql.{Connection, PreparedStatement}
import java.text.SimpleDateFormat
import java.util.Date

object Spark_Click_Rank {
  def main(args: Array[String]): Unit = {
    val ssc: StreamingContext = new StreamingContext(new SparkConf().setMaster("local[*]")
      .setAppName("spark-streaming")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .set("spark.testing.memory", "2147480000"), Duration.apply(3000))

    val kafkaPara: Map[String, Object] = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> PropUtil("kafka.brokers"),
      ConsumerConfig.GROUP_ID_CONFIG -> PropUtil("kafka.group"),
      "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
    )
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils
      .createDirectStream[String, String](ssc, LocationStrategies.PreferConsistent,
        ConsumerStrategies.Subscribe[String, String](Set(PropUtil("kafka.topic")), kafkaPara))
    kafkaDStream.map(x => {
      val str: String = x.value()
      val strs: Array[String] = str.split(" ")
      // timestamp area city userid adid
      val dayTs: Long = strs(0).toLong / 1000 / 60 / 60 / 24 * 24 * 60 * 60 * 1000
      ((dayTs, strs(1), strs(2), strs(4)), 1)
    }).reduceByKey(_ + _).foreachRDD(
      rdd => {
        rdd.foreachPartition(iter => {
          val conn: Connection = DruidUtil.getConnection
          iter.toArray.foreach(
            item => {
              // ts area city adid
              val stat: PreparedStatement = conn.prepareStatement(
                """
                  |insert into
                  |area_city_ad_count
                  |values (?,?,?,?,?)
                  |ON DUPLICATE KEY UPDATE
                  | count = count + ?
                  |""".stripMargin)
              stat.setString(1, new SimpleDateFormat("yyyy-MM-dd").format(new Date(item._1._1)))
              stat.setString(2, item._1._2)
              stat.setString(3, item._1._3)
              stat.setString(4, item._1._4)
              stat.setInt(5, item._2)
              stat.setInt(6, item._2)
              stat.executeUpdate()
            }
          )
          conn.close()


        })

      }
    )
    ssc.start()
    ssc.awaitTermination()
  }
}
